package com.yifeng.repo.flink.data.transport.utils.executors;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;


/**
 * @Date 2020/6/4
 * @Author wang zhi
 */
public class FireExecutor {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(FireExecutor.class);

    private ExecutorService executorService;

    public FireExecutor(DataSourceConfig dataSourceConfig) {
    	if(dataSourceConfig.getParallelism() <= 0) {
    		return;
    	}
		if(dataSourceConfig.getParallelism() == 1) {
			ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("full-task" + dataSourceConfig.getTableList().get(0)).build();
	        this.executorService = Executors.newSingleThreadExecutor(threadFactory);
		    return;
		}
		this.executorService = newThreadExecutor(dataSourceConfig.getParallelism(),dataSourceConfig.getTableList().size(),"full-task");
    }

    private ExecutorService newThreadExecutor(Integer corePoolSize, Integer maximumPoolSize, String threadNamePrefix) {
        return new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                60L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(maximumPoolSize),
                new ThreadFactory() {
                    final AtomicInteger threadNumber = new AtomicInteger(0);
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName(threadNamePrefix + threadNumber.incrementAndGet());
                        thread.setDaemon(true);
                        return thread;
                    }
                },
                new ThreadPoolExecutor.AbortPolicy());
    }


    public void shutdown() {
        this.executorService.shutdown();
    }
    
    public void shutdownNow() {
    	if(this.executorService != null) {
    		this.executorService.shutdownNow();
    	}
    }

    public boolean isShutdown() {
       return this.executorService.isShutdown();
    }
    
    public boolean isTerminated() {
        return this.executorService.isTerminated();
    }

    public <T> void submitTask(Callable<T> task) {
        this.executorService.submit(task);
    }

    public <T> void submitTask(Runnable task) {
        this.executorService.submit(task);
    }
    
    public <T> void batchSubmitTask(Collection<? extends Callable<T>> tasks) {
        for (Callable<T> task : tasks) {
            this.executorService.submit(task);
        }
    }


    public <T> T runTask(Callable<T> task) {
        Future<T> future = this.executorService.submit(task);
        try {
            return future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    public <T> Future<T> runReturnFuture(Callable<T> task) {
        return this.executorService.submit(task);
    }
    
    public boolean awaitTermination(long timeout, TimeUnit unit) {
    	if(this.executorService == null) {
    		return true;
    	}
    	try {
			return this.executorService.awaitTermination(timeout,unit);
		} catch (InterruptedException e) {
			LOGGER.error("关闭线程池失败");
		}
    	return false;
    }

}
